﻿using LJ.Abstractions.DependencyInjection;
using LJ.StreamSystem.LazyHare.Transport.Abstractions.Exceptions;
using Microsoft.Extensions.DependencyInjection;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace LJ.StreamSystem.LazyHare.Transport.Kafka.Internal;

public class KafkaStreamSystemClientFactory(IDependencyInjectionFactory dependencyInjectionFactory) : IStreamSystemClientFactory
{
    private readonly IDependencyInjectionFactory _dependencyInjectionFactory = dependencyInjectionFactory;

    public IStreamSystemClient Create(string applicationId, string aggregateSummaryTopic, string[] aggregateTopics)
    {
        try
        {
            using var serviceScope = _dependencyInjectionFactory.CreateScope();
            var serviceProvider = serviceScope.ServiceProvider;

            var streamSystemClient = serviceProvider.GetRequiredService<IStreamSystemClient>();

            if (streamSystemClient is not KafkaStreamSystemClient kafkaStreamSystemClient)
            {
                throw new ArgumentException($"{GetType().FullName} IStreamSystemClient is not {nameof(kafkaStreamSystemClient)}");
            }
            kafkaStreamSystemClient.Init(applicationId, aggregateSummaryTopic, aggregateTopics);
            return kafkaStreamSystemClient;

        }
        catch (Exception e)
        {
            throw new StreamSystemBrokerConnectionException(e);
        }
    }
}
